package cn.hesion.mergefile.step1;
/**
 * ClassName: MergeRecordReader <br/>
 * Description: <br/>
 * date: 2021/2/4 9:34<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import javax.xml.stream.FactoryConfigurationError;
import java.io.IOException;
import java.time.temporal.ValueRange;

/**
 * @program: ClientDemo
 * @description:
 * @author: hesion
 * @create: 2021-02-04 09:34
 **/
public class MergeRecordReader extends RecordReader<Text, BytesWritable> {
    private FileSplit split;
    private Configuration conf;

    Text k = new Text();
    BytesWritable v = new BytesWritable();

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        this.split = (FileSplit) split;
        this.conf = context.getConfiguration();
    }
private boolean flag = true;
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if(flag){
            //准备一个数组存放读取到的数据
            byte[] content = new byte[((int) split.getLength())];
            // 获取切片的path
            Path path = split.getPath();
            // 获取文件系统
            FileSystem fs = path.getFileSystem(conf);

            FSDataInputStream fis = fs.open(path);

            IOUtils.readFully(fis,content,0,content.length);

            k.set(path.toString());
            v.set(content,0,content.length);

            IOUtils.closeStream(fis);

            flag=false;
            return true;
        }
        return false;
    }

    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return k;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return v;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {
    }
}
